package cn.doitedu.sql;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/4/12
 * @Desc: 学大数据，上多易教育
 *   mysql-cdc连接器使用测试
 **/
public class Demo02_MySql_CDC_Connector {
    public static void main(String[] args) {


        // 构建编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);


        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 用连接器建表，映射物理数据源
        tenv.executeSql(
                        " create table stu_info_cdc(         "+
                        "    id int,                         "+
                        "    name string,                    "+
                        "    gender string,                  "+
                        "    age int,                        "+
                        "    score float,                    "+
                        "    primary key(id) not enforced    "+
                        " ) WITH (                           "+
                        "     'connector' = 'mysql-cdc',     "+
                        "     'hostname' = 'doitedu',        "+
                        "     'port' = '3306',               "+
                        "     'username' = 'root',           "+
                        "     'password' = 'root',           "+
                        "     'database-name' = 'doit47',    "+
                        "     'table-name' = 'stu_info'      "+
                        " )                                  "
        );



        // 写sql做运算
        tenv.executeSql("select gender,avg(score) as avg_score from stu_info_cdc group by gender").print();


    }
}
